package com.sy.jtamqproducer.listener;

import java.util.Optional;

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import com.alibaba.fastjson.JSONObject;
import com.sy.jtamqproducer.dao.TraLogReposity;
import com.sy.jtamqproducer.pojo.TestS2;
import com.sy.jtamqproducer.pojo.TraLog;
import com.sy.jtamqproducer.service.TestService;

@RocketMQTransactionListener
public class TestTransactionListener implements RocketMQLocalTransactionListener {
	private Logger logger = LoggerFactory.getLogger(TestTransactionListener.class);
	@Autowired
	private TestService testService;
	@Autowired
	private TraLogReposity traLogReposity;

    
	/**
     * 回查
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
    	MessageHeaders headers = message.getHeaders();
        //获取事务ID
        String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
        logger.info("检查本地事务,事务ID:{}",transactionId);
        //根据事务id从日志表检索
        Optional<TraLog> traLogOpt = traLogReposity.findById(transactionId);
        if(traLogOpt.isPresent()){
            return RocketMQLocalTransactionState.COMMIT;
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }

    /**
     * 发送half msg 返回send ok后调用的方法
     * 发送半消息(half)
     * 
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg) {
        logger.info("开始执行本地事务....");
        MessageHeaders headers = message.getHeaders();
        //获取事务ID
        String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
        String testS2Str = (String) headers.get("body");
        TestS2 testS2 = JSONObject.toJavaObject(JSONObject.parseObject(testS2Str), TestS2.class);
        logger.info("transactionId is {}, TestS2 is {}",transactionId, testS2Str);
        try{
            //执行本地事务，并记录日志
        	testService.saveJtaTest(testS2, transactionId);
            //执行成功，可以提交事务
            return RocketMQLocalTransactionState.COMMIT;
        }catch (Exception e){
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }


}